package org.budo.redis.jdbc.driver.executor.socket;

import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;

import org.budo.support.assertion.Assert;
import org.budo.support.java.net.util.SocketUtil;
import org.budo.support.lang.util.IoUtil;

/**
 * @author lmw
 */
public class BudoSocketRedisStatementTransporterImpl implements BudoSocketRedisStatementTransporter {
    private static final byte RESPONSE_TYPE_STRING = '+', //
            RESPONSE_TYPE_INTEGER = ':', //
            RESPONSE_TYPE_ARRAY = '*', //
            RESPONSE_TYPE_BULK_STRING = '$', //
            RESPONSE_TYPE_ERROR = '-';

    private String host;

    private int port;

    private Socket socket;

    private BudoRedisSocketInputStream redisInputStream;

    private OutputStream outputStream;

    public BudoSocketRedisStatementTransporterImpl(String host, int port) {
        this.host = host;
        this.port = port;

        this.socket = SocketUtil.newSocket(host, port);

        this.outputStream = SocketUtil.getOutputStream(socket);
        InputStream inputStream = SocketUtil.getInputStream(socket);

        this.redisInputStream = new BudoRedisSocketInputStream(inputStream);
    }

    @Override
    public void writeAndFlush(byte[] command) {
        Assert.notNull(command, "#140 command can not be null");

        try {
            this.outputStream.write(command);
            this.outputStream.flush();
        } catch (Throwable e) {
            throw new IllegalStateException("#146 error=" + e + ", command=" + new String(command), e);
        }
    }

    @Override
    public Object readResponse() {
        byte responseType = redisInputStream.readByte();

        if (RESPONSE_TYPE_STRING == responseType || RESPONSE_TYPE_INTEGER == responseType) {
            return redisInputStream.readString();
        }

        if (RESPONSE_TYPE_BULK_STRING == responseType) {
            return redisInputStream.readBulkString();
        }

        if (RESPONSE_TYPE_ARRAY == responseType) {
            return this.readArray();
        }

        if (RESPONSE_TYPE_ERROR == responseType) {
            byte[] message = redisInputStream.readString();
            throw new IllegalStateException("#47 " + new String(message) + ", socket response error, this=" + this);
        }

        throw new IllegalStateException("#50 readResponse error, not supported responseType, " + (char) responseType);
    }

    private byte[][] readArray() {
        int arrayLen = (int) redisInputStream.readCRLF();

        if (arrayLen == -1) {
            return null;
        }

        byte[][] array = new byte[arrayLen][];
        for (int i = 0; i < arrayLen; i++) {
            Object value = this.readResponse();
            array[i] = (byte[]) value;
        }

        return array;
    }

    public void close() {
        IoUtil.close(this.outputStream);
        IoUtil.close(this.redisInputStream);

        SocketUtil.close(this.socket);
    }

    @Override
    public String toString() {
        return super.toString() //
                + ", host=" + this.host //
                + ", port=" + this.port //
                + ", socket=" + this.socket;
    }
}